package mr.weblog.part1;

import java.io.IOException;

import mr.weblog.support.TextPair;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class StockDataMapper extends Mapper<LongWritable, Text, TextPair, TextPair> {
	private static Logger log = LoggerFactory.getLogger(StockDataMapper.class);

	public final static String delim = ",";

	private final Text indicator = new Text("R");
	private final Text joinKey = new Text();
	private final Text joinVal = new Text();

	private final TextPair joinKeyPair = new TextPair();
	private final TextPair valuePair = new TextPair();
	{
		joinKeyPair.setSide(indicator);
		valuePair.setSide(indicator);
	}

    @Override
    public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String [] split = value.toString().split(delim);
        if(split.length != 3) {
        	log.warn("bad stock price input: " + value.toString());
			context.getCounter("Assig4", "Bad Stock Data Entry").increment(1);
        }
        else{
	        joinKey.set(split[0]);
			joinKeyPair.setFirst(joinKey);
			
			joinVal.set(split[1] + delim + split[2]);
			valuePair.setFirst(joinVal);
			context.write(joinKeyPair, valuePair);
        }
    }
}